package avicit.bdp.dcs.task.service;

import avicit.bdp.common.utils.BdpLogUtil;
import avicit.bdp.common.utils.CheckParaUtils;
import avicit.bdp.common.utils.database.ColumnInfo;
import avicit.bdp.common.utils.enums.JobType;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dcs.datasource.dto.JobDatasource;
import avicit.bdp.dcs.datasource.service.BaseDbService;
import avicit.bdp.dcs.datasource.service.DataSourceUtils;
import avicit.bdp.dcs.datasource.service.DsDatasourceService;
import avicit.bdp.dcs.datasource.service.TableService;
import avicit.bdp.dcs.detail.dto.TaskDataDetailDTO;
import avicit.bdp.dcs.detail.dto.TaskTableDetailDTO;
import avicit.bdp.dcs.detail.service.TaskDataDetailService;
import avicit.bdp.dcs.detail.service.TaskTableDetailService;
import avicit.bdp.dcs.job.dto.JobDTO;
import avicit.bdp.dcs.job.service.DcsJobService;
import avicit.bdp.dcs.json.service.DataxJsonService;
import avicit.bdp.dcs.task.dto.ColumnMappingDto;
import avicit.bdp.dcs.task.dto.TableMappingDto;
import avicit.bdp.dcs.task.dto.TaskDTO;
import avicit.bdp.dcs.tools.enums.Enable;
import avicit.bdp.dcs.tools.enums.ProcessMode;
import avicit.bdp.dcs.tools.enums.SyncCondition;
import avicit.bdp.dcs.tools.enums.SyncMode;
import avicit.bdp.dcs.utils.DcsErrorCode;
import avicit.platform6.commons.utils.ComUtil;
import avicit.platform6.core.exception.BusinessException;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;
import java.util.stream.Collectors;

/**
 * @金航数码科技有限责任公司
 * @作者：liyb
 * @邮箱：liyb@avic-digital.com
 * @创建时间： 2020-09-08 15:36
 * @类说明： 结构化任务服务类
 * @修改记录：
 */
@Service
@Transactional(rollbackFor = Exception.class)
public class StructureTaskService implements TaskInterface {

  private static final Logger logger = LoggerFactory.getLogger(StructureTaskService.class);

  @Autowired private TaskService taskService;
  @Autowired private DcsJobService jobService;
  @Autowired private DsDatasourceService dsDatasourceService;
  @Autowired private DataxJsonService dataxJsonService;
  @Autowired private TableService tableService;
  @Autowired private TaskTableDetailService taskTableDetailService;
  @Autowired private TaskDataDetailService taskDataDetailService;
  @Autowired private DataSourceUtils dataSourceUtils;
  @Autowired private BaseDbService baseDbService;

  /** 查看任务 */
  @Override
  public TaskDTO getTaskById(JobDTO jobDTO, String srcTableName) throws Exception {
    // 1、查询任务
    TaskDTO query = new TaskDTO();
    query.setJobId(jobDTO.getId());
    query.setSrcTableName(srcTableName);
    query.setFlag(Enable.ZERO.getCode());
    TaskDTO task = taskService.selectOne(query);
    CheckParaUtils.checkDbObjectValidity(
        task,
        logger,
        String.format("没有查到数据，请确保jobId：%s,srcTableName:%s的任务是否存在", jobDTO.getId(), srcTableName));
    task.setDataxJson(null);

    // 2、查询源表列字段
    List<ColumnInfo> srcColumnList =
        baseDbService.getColumnList(jobDTO.getSrcDatasourceId(), srcTableName);
    task.setSrcColumnList(srcColumnList);

    // 3、获取列映射关系
    Map<String, String> columnMap = new HashMap<>(16);
    List<ColumnMappingDto> columnMappingDtoList =
        JSONUtils.toList(task.getMappingColumn(), ColumnMappingDto.class);
    if (CollectionUtils.isNotEmpty(columnMappingDtoList)) {
      for (ColumnMappingDto columnMappingDto : columnMappingDtoList) {
        columnMap.put(columnMappingDto.getSrcColumnName(), columnMappingDto.getDstColumnName());
      }
    }

    // 4、查询目标表列字段
    List<ColumnInfo> lastColumnList = new ArrayList<>();
    List<ColumnInfo> dstColumnList =
        baseDbService.getColumnList(jobDTO.getDstDatasourceId(), task.getDstTableName());
    if (CollectionUtils.isNotEmpty(dstColumnList)) {
      for (ColumnInfo columnInfo : dstColumnList) {
        ColumnInfo dstColumnInfo = new ColumnInfo();
        String modifyColumnName = columnMap.get(columnInfo.getName());
        dstColumnInfo.setName(modifyColumnName == null ? columnInfo.getName() : modifyColumnName);
        //                dstColumnInfo.setName(dstColumnInfo.getName().toLowerCase());
        dstColumnInfo.setName(dstColumnInfo.getName());
        dstColumnInfo.setType(columnInfo.getType());
        dstColumnInfo.setLength(columnInfo.getLength());
        lastColumnList.add(dstColumnInfo);
      }
    } else {
      for (ColumnInfo columnInfo : task.getSrcColumnList()) {
        ColumnInfo dstColumnInfo = new ColumnInfo();
        String modifyColumnName = columnMap.get(columnInfo.getName());
        dstColumnInfo.setName(modifyColumnName == null ? columnInfo.getName() : modifyColumnName);
        dstColumnInfo.setType(columnInfo.getType());
        dstColumnInfo.setLength(columnInfo.getLength());
        lastColumnList.add(dstColumnInfo);
      }
    }

    lastColumnList =
        taskService.sortBySrcColumn(srcColumnList, lastColumnList, task.getMappingColumn());
    task.setDstColumnList(lastColumnList);

    // 5、查询时间列字段
    task.setTimeColumnList(dsDatasourceService.getDateTimeColumnList(task.getId()));

    // 6、查询主键列字段
    task.setPrimaryKeyColumnList(dsDatasourceService.getPrimaryKeyList(task.getId()));
    return task;
  }

  /**
   * 验证数据
   *
   * @param oldTask
   * @param taskDTO
   * @return
   * @throws Exception
   */
  private void checkData(TaskDTO oldTask, TaskDTO taskDTO) throws Exception {
    JobDTO jobDTO = jobService.selectByPrimaryKey(oldTask.getJobId());
    CheckParaUtils.checkDbObjectValidity(
        jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), oldTask.getJobId()));
    CheckParaUtils.checkStringValidity(taskDTO.getDstTableName(), logger, "目标表名必填");
    CheckParaUtils.checkObjectValidity(taskDTO.getModifyColumn(), logger, "是否修改列映射必填");
  }

  /**
   * 判断列名称合法性
   *
   * @param taskDTO
   * @throws Exception
   */
  private void checkColumnName(TaskDTO taskDTO) throws Exception {
    if (!StringUtils.isEmpty(taskDTO.getMappingColumn())) {
      List<ColumnMappingDto> columnMappingDtos =
          JSONUtils.toList(taskDTO.getMappingColumn(), ColumnMappingDto.class);
      this.checkColumnName(columnMappingDtos);
    }
  }

  /**
   * 验证过滤条件
   *
   * @param taskDTO
   * @throws Exception
   */
  private void checkGrammar(TaskDTO taskDTO) throws Exception {
    if (!this.checkGrammar(taskDTO.getId(), taskDTO.getFilterCondition())) {
      throw new BusinessException("语法验证不通过,请输入正确的过滤条件");
    }
  }

  /**
   * 验证如果选择的是增量同步，且没有选择时间戳字段
   *
   * @throws Exception
   */
  public void checkSync(Integer jobType, Integer syncMode, String fieldName, Date syncTimePoint, String srcTableName) throws Exception {
    if (jobType == JobType.SYNC.getCode()
        && syncMode == SyncMode.DATAX_INCREMENT_SYNC.getCode()) {
      if (StringUtils.isEmpty(fieldName) || syncTimePoint == null) {
        throw new BusinessException(String.format("表{}是增量同步,必须选择时间戳字段并且设置同步开始时间", srcTableName));
      }
    }

    if (jobType == JobType.SYNC.getCode()
            && syncMode == SyncMode.COMMON_INCREMENT_SYNC.getCode()) {
      if (StringUtils.isEmpty(fieldName)) {
        throw new BusinessException("唯一约束和增量字段必填");
      }

      String uniqueKey = null;
      String incrementalColumn = null;
      try{
        JSONObject json = JSONObject.parseObject(fieldName);
        uniqueKey = (String) json.get("uniqueKey");
        incrementalColumn = (String) json.get("incrementalColumn");
      }catch (Exception e){
        logger.error("解析增量字段报错,{}", e.getMessage());
      }

      if (StringUtils.isEmpty(uniqueKey)) {
        throw new BusinessException("唯一约束字段必填");
      }
      if (StringUtils.isEmpty(incrementalColumn)) {
        throw new BusinessException("增量字段必填");
      }
      if(uniqueKey.equals(incrementalColumn)){
        throw new BusinessException("唯一约束和增量字段不能是同一个字段");
      }
    }
  }

  /** 更新任务 */
  @Override
  public String updateTask(TaskDTO oldTask, TaskDTO taskDTO) throws Exception {
    // 验证数据
    checkData(oldTask, taskDTO);

    // 判断列名称合法性
    checkColumnName(taskDTO);

    // 验证过滤条件
    checkGrammar(taskDTO);

    // 验证如果选择的是增量同步，且没有选择时间戳字段
    checkSync(oldTask.getJobType(), oldTask.getSyncMode(), taskDTO.getFieldName(), taskDTO.getSyncTimePoint(), oldTask.getSrcTableName());

    // 记录日志
    BdpLogUtil.log4Update(taskDTO, oldTask);

    // 先更新列映射关系，再生成dataxjson
    taskService.updateByPrimaryKeySelective(taskDTO);
    return dataxJsonService.buildDataxJson(oldTask);
  }

  /**
   * 新增task
   *
   * @param jobDTO
   * @param tableMappingDto
   * @throws Exception
   */
  private TaskDTO insertTask(JobDTO jobDTO, TableMappingDto tableMappingDto) throws Exception {
    String mappingColumnString =
        this.getMappingColumnString(
            tableMappingDto.getSrc_table(),
            tableMappingDto.getDst_table(),
            jobDTO.getSrcDatasourceId(),
            jobDTO.getDstDatasourceId());
    TaskDTO taskDTO = new TaskDTO();
    taskDTO.setId(ComUtil.getId());
    taskDTO.setJobId(jobDTO.getId());
    taskDTO.setJobType(jobDTO.getJobType());
    taskDTO.setSrcTableName(tableMappingDto.getSrc_table());
    taskDTO.setDstTableName(tableMappingDto.getDst_table());
    taskDTO.setMappingColumn(mappingColumnString);
    taskDTO.setSyncMode(jobDTO.getSyncMode());
    taskDTO.setIsCreated(Enable.ZERO.getCode());
    taskDTO.setFlag(Enable.ZERO.getCode());
    taskService.insert(taskDTO);

    // 记录日志
    BdpLogUtil.log4Insert(taskDTO);
    logger.info("新增任务成功,dstTable:{}", tableMappingDto.getDst_table());

    return taskDTO;
  }

  /**
   * 更新task
   *
   * @param jobDTO
   * @param tableMappingDto
   * @throws Exception
   */
  private TaskDTO updateTask(JobDTO jobDTO, TableMappingDto tableMappingDto) throws Exception {
    TaskDTO query = new TaskDTO();
    query.setJobId(jobDTO.getId());
    query.setSrcTableName(tableMappingDto.getSrc_table());
    query.setFlag(Enable.ZERO.getCode());
    TaskDTO taskDTO = taskService.selectOne(query);

    // 目标表发生改变需要更新数据
    if (!taskDTO.getDstTableName().equals(tableMappingDto.getDst_table())) {
      String mappingColumnString =
          this.getMappingColumnString(
              tableMappingDto.getSrc_table(),
              tableMappingDto.getDst_table(),
              jobDTO.getSrcDatasourceId(),
              jobDTO.getDstDatasourceId());
      taskDTO.setMappingColumn(mappingColumnString);
      taskDTO.setDstTableName(tableMappingDto.getDst_table());
      taskDTO.setFilterCondition(null);
      taskDTO.setModifyColumn(Enable.ZERO.getCode());
      taskDTO.setUpdateHistoryData(Enable.ONE.getCode());
      taskDTO.setSyncConditionField(SyncCondition.TIME.getCode());
      taskDTO.setFieldName(null);
      taskDTO.setSyncTimePoint(null);
      taskDTO.setSyncIdValue(null);
    }

    if (taskDTO.getSyncMode() != null && !taskDTO.getSyncMode().equals(jobDTO.getSyncMode())) {
      taskDTO.setFieldName(null);
    }

    taskDTO.setSyncMode(jobDTO.getSyncMode());
    taskDTO.setIsCreated(Enable.ZERO.getCode());
    taskService.updateByPrimaryKey(taskDTO);
    logger.info("修改任务成功,dstTable:{}", tableMappingDto.getDst_table());

    // 删除上次任务保存bdp_dcs_task_table_detail表数据和保存bdp_dcs_task_data_detail表数据
    TaskTableDetailDTO taskTableDetailDTO =
        TaskTableDetailDTO.builder().taskId(taskDTO.getId()).build();
    taskTableDetailService.delete(taskTableDetailDTO);

    TaskDataDetailDTO taskDataDetailDTO =
        TaskDataDetailDTO.builder().taskId(taskDTO.getId()).build();
    taskDataDetailService.delete(taskDataDetailDTO);
    return taskDTO;
  }

  /**
   * 删除任务
   *
   * @param jobDTO
   * @param srcTableList
   * @throws Exception
   */
  private void deleteTask(JobDTO jobDTO, List<String> srcTableList) throws Exception {
    // 根据jobId查询taskList，和selectedTable比较，selectedTable里边不存在的srcTable，就删除task，同时删除垃圾数据
    List<TableMappingDto> tableMappingDtoList =
        JSONUtils.toList(jobDTO.getSelectedTable(), TableMappingDto.class);
    List<String> srcTableMappingList =
        tableMappingDtoList.stream()
            .map(TableMappingDto::getSrc_table)
            .collect(Collectors.toList());
    List<String> collect =
        srcTableList.stream()
            .filter(s -> !srcTableMappingList.contains(s))
            .collect(Collectors.toList());
    collect.forEach(
        srcTableName -> {
          TaskDTO taskDtoQuery = new TaskDTO();
          taskDtoQuery.setJobId(jobDTO.getId());
          taskDtoQuery.setSrcTableName(srcTableName);
          taskDtoQuery.setFlag(Enable.ZERO.getCode());
          List<TaskDTO> taskList = taskService.selectList(taskDtoQuery);
          taskList.forEach(
              task -> {
                TaskTableDetailDTO taskTableDetailDTO = new TaskTableDetailDTO();
                taskTableDetailDTO.setTaskId(task.getId());
                taskTableDetailService.delete(taskTableDetailDTO);
                TaskDataDetailDTO taskDataDetailDTO = new TaskDataDetailDTO();
                taskDataDetailDTO.setTaskId(task.getId());
                taskDataDetailService.delete(taskDataDetailDTO);

                // 删除任务
                task.setFlag(1);
                taskService.updateByPrimaryKey(task);
              });
          //            taskService.delete(taskDtoQuery);
          logger.info("删除任务成功,srcTableName:{}", srcTableName);
        });
  }

  /** 批量保存任务 一个job关联多个task */
  @Override
  public void saveTasksByJobId(JobDTO jobDTO) throws Exception {
    List<String> srcTableList = taskService.getSrcTableListByJobId(jobDTO.getId());

    // 根据jobId和srcTable查询task存不存在，存在就更新，不存在新增
    List<TableMappingDto> tableMappingDtoList =
        JSONUtils.toList(jobDTO.getSelectedTable(), TableMappingDto.class);

    for (TableMappingDto tableMappingDto : tableMappingDtoList) {
      TaskDTO taskDTO = null;
      if (!srcTableList.contains(tableMappingDto.getSrc_table())) {
        taskDTO = insertTask(jobDTO, tableMappingDto);
      } else {
        taskDTO = updateTask(jobDTO, tableMappingDto);
      }

      // 保存bdp_dcs_task_table_detail表数据和保存bdp_dcs_task_data_detail表数据
      taskService.saveTaskDetailData(
          tableMappingDto.getSrc_table(), tableMappingDto.getDst_table(), taskDTO, jobDTO);
    }

    // 删除任务
    deleteTask(jobDTO, srcTableList);
  }

  /**
   * 创建目标表并回写建表sql语句
   *
   * @param jobDTO jobDTO
   * @param taskDTO taskDTO
   * @return 创建目标表结果
   * @throws Exception
   */
  @Override
  public Boolean createTable(JobDTO jobDTO, TaskDTO taskDTO) throws Exception {
    logger.info("processMode:{}", jobDTO.getProcessMode());
    boolean createTable = true;

    if (jobDTO.getJobType() == JobType.SYNC.getCode()) {
      if (taskDTO.getSyncMode() == SyncMode.TOTAL_SYNC.getCode()) {
        createTable = totalSyncPreHandle(jobDTO, taskDTO);
      } else {
        createTable = incrementSyncPreHandle(jobDTO, taskDTO);
      }
    } else if (jobDTO.getJobType() == JobType.TRANSFER.getCode()) {
      createTable = totalSyncPreHandle(jobDTO, taskDTO);
    } else {
      logger.warn("不支持的类型{}", jobDTO.getJobType());
      return false;
    }

    if (!createTable) {
      return true;
    }

    // 列名称转换
    Map<String, String> columnMap = taskService.convertColumnName(taskDTO);
    Map<String, String> map =
        baseDbService.createTable(
            jobDTO, taskDTO.getSrcTableName(), taskDTO.getDstTableName(), columnMap);
    return taskService.updateSql(map, taskDTO);
  }

  private Boolean incrementSyncPreHandle(JobDTO jobDTO, TaskDTO taskDTO) throws Exception {
    // 增量同步，第一次需要删除目标表，第二次及以后不能删除目标表
    // 根据是否有历史任务，判断是否为第一次执行job
    //        boolean isCoverTableFlag = taskService.isCoverTable(jobDTO, taskDTO);
    //        logger.info("isCoverTableFlag:{}", isCoverTableFlag);

    boolean isCoverTableFlag = true;
    if (taskDTO.getIsCreated() != null && Enable.ONE.getCode() == taskDTO.getIsCreated()) {
      logger.info("表已经被创建过了:{}", taskDTO.getDstTableName());
      isCoverTableFlag = false;
    }
    logger.info("isCoverTableFlag:{}", isCoverTableFlag);

    boolean existDstTable =
        taskService.isExistTable(jobDTO.getDstDatasourceId(), taskDTO.getDstTableName());
    logger.info("existDstTable:{}", existDstTable);
    taskService.saveTableLog(taskDTO, String.format("目标库存在目标表:%s", existDstTable));

    if (existDstTable && isCoverTableFlag) {
      if (jobDTO.getProcessMode() == ProcessMode.COVER_DST_TABLE.getCode()) {
        boolean isDel =
            baseDbService.dropTable(jobDTO.getDstDatasourceId(), taskDTO.getDstTableName());
        logger.info("删除目标表:{},删除结果:{}", taskDTO.getDstTableName(), isDel);
        taskService.saveTableLog(
            taskDTO, String.format("删除目标表:%s,删除结果:%s", taskDTO.getDstTableName(), isDel));
      } else if (jobDTO.getProcessMode() == ProcessMode.RETAIN_DST_TABLE.getCode()) {
        logger.info("已存在表的处理模式为保留目标表，不做处理");
        return false;
      } else if (jobDTO.getProcessMode() == ProcessMode.CLEAR_DST_TABLE_DATA.getCode()) {
        boolean isClear =
            baseDbService.clearTable(jobDTO.getDstDatasourceId(), taskDTO.getDstTableName());
        logger.info("清空目标表数据:{},清空结果:{}", taskDTO.getDstTableName(), isClear);
        return false;
      } else {
        throw new BusinessException("不支持的已存在表的处理模式");
      }
    }

    return isCoverTableFlag;
  }

  private Boolean totalSyncPreHandle(JobDTO jobDTO, TaskDTO taskDTO) throws Exception {
    boolean existDstTable =
        taskService.isExistTable(jobDTO.getDstDatasourceId(), taskDTO.getDstTableName());
    logger.info("existDstTable:{}", existDstTable);
    taskService.saveTableLog(taskDTO, String.format("目标库存在目标表:%s", existDstTable));

    // 目标表不存在，直接新建表
    if (existDstTable) {
      // 目标已存在表的处理模式，预检查并报错拦截：0，覆盖目标表：1，保留目标表：2
      if (jobDTO.getProcessMode() == ProcessMode.PRECHECK_ERROR.getCode()) {
        JobDatasource dstDatasource =
            dataSourceUtils.convertDatasource(jobDTO.getDstDatasourceId());
        logger.info(
            DcsErrorCode.DST_DB_EXIST_SAME_TABLE.getDesc(),
            dstDatasource.getJdbcUrl(),
            taskDTO.getDstTableName());
        throw new BusinessException(
            String.format(
                DcsErrorCode.DST_DB_EXIST_SAME_TABLE.getDesc(),
                dstDatasource.getJdbcUrl(),
                taskDTO.getDstTableName()));
      } else if (jobDTO.getProcessMode() == ProcessMode.COVER_DST_TABLE.getCode()) {
        boolean isDel =
            baseDbService.dropTable(jobDTO.getDstDatasourceId(), taskDTO.getDstTableName());
        logger.info("删除目标表:{},删除结果:{}", taskDTO.getDstTableName(), isDel);
        taskService.saveTableLog(
            taskDTO, String.format("删除目标表:%s,删除结果:%s", taskDTO.getDstTableName(), isDel));
      } else if (jobDTO.getProcessMode() == ProcessMode.RETAIN_DST_TABLE.getCode()) {
        logger.info("已存在表的处理模式为保留目标表，不做处理");
        return false;
      } else if (jobDTO.getProcessMode() == ProcessMode.CLEAR_DST_TABLE_DATA.getCode()) {
        boolean isClear =
            baseDbService.clearTable(jobDTO.getDstDatasourceId(), taskDTO.getDstTableName());
        logger.info("清空目标表数据:{},清空结果:{}", taskDTO.getDstTableName(), isClear);
        return false;
      } else {
        throw new BusinessException("不支持的已存在表的处理模式");
      }
    }

    return true;
  }

  /**
   * 获取列映射关系
   *
   * @param srcTable srcTable
   * @param dstTable dstTable
   * @param srcDatasourceId srcDatasourceId
   * @param dstDatasourceId dstDatasourceId
   * @return 结构化mapping columns
   * @throws Exception e
   */
  private String getMappingColumnString(
      String srcTable, String dstTable, String srcDatasourceId, String dstDatasourceId)
      throws Exception {
    Boolean isExistDstTable = taskService.isExistInDstDatabase(dstDatasourceId, dstTable);
    ArrayList<ColumnMappingDto> columnMappingDtoList;
    if (isExistDstTable) {
      // 目标数据库存在目标表
      // 获取源表列名称以及类型
      List<ColumnInfo> srcColumnList = baseDbService.getColumnList(srcDatasourceId, srcTable);

      // 获取目标表列名称以及类型
      List<ColumnInfo> dstColumnList = baseDbService.getColumnList(dstDatasourceId, dstTable);
      logger.info("srcColumnList:{}", JSONArray.toJSONString(srcColumnList));
      logger.info("dstColumnList:{}", JSONArray.toJSONString(dstColumnList));
      columnMappingDtoList = taskService.getMergeColumnList(srcColumnList, dstColumnList);
    } else {
      // 目标数据库不存在目标表
      // 获取源表列名称以及类型
      List<ColumnInfo> mergeColumnList = baseDbService.getColumnList(srcDatasourceId, srcTable);
      JobDatasource dstDatasource = dataSourceUtils.convertDatasource(dstDatasourceId);
      columnMappingDtoList =
          taskService.getColumnMappingDtos(mergeColumnList, dstDatasource.getDbType());
    }

    logger.info("列映射关系为:{}", JSONArray.toJSONString(columnMappingDtoList));
    return JSONArray.toJSONString(columnMappingDtoList);
  }

  /** 检查列名合法性 */
  private void checkColumnName(List<ColumnMappingDto> columnMappingDtos) {
    for (ColumnMappingDto columnMappingDto : columnMappingDtos) {
      if (!tableService.isValidColumnName(columnMappingDto.getDstColumnName())) {
        throw new BusinessException(
            String.format(
                DcsErrorCode.COLUMN_NAME_CHECK.getDesc(), columnMappingDto.getDstColumnName()));
      }
    }
  }

  /** 验证语法 */
  @Transactional
  public Boolean checkGrammar(String taskId, String filterCondition) throws Exception {
    // 过滤条件为空直接返回true
    if (StringUtils.isEmpty(filterCondition)) {
      return true;
    }

    TaskDTO taskDTO = taskService.selectByPrimaryKey(taskId);
    CheckParaUtils.checkDbObjectValidity(
        taskDTO, logger, String.format(DcsErrorCode.TASK_NULL.getDesc(), taskId));
    JobDTO jobDTO = jobService.selectByPrimaryKey(taskDTO.getJobId());
    CheckParaUtils.checkDbObjectValidity(
        jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), taskDTO.getJobId()));

    // 语法验证
    boolean success =
        baseDbService.checkGrammar(
            jobDTO.getSrcDatasourceId(), taskDTO.getSrcTableName(), filterCondition);
    logger.info(
        String.format(
            "语法验证结果：%s,数据源id：%s,表名：%s,过滤条件：%s",
            success, jobDTO.getSrcDatasourceId(), taskDTO.getSrcTableName(), filterCondition));
    return success;
  }

  @Override
  public Integer getType() {
    return JobType.TRANSFER.getCode();
  }
}
